package olap4cloud.hbase;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.List;

import olap4cloud.data.DataStruct;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;

public class TrackingTable {

	public static final String tableName = "Tracking";
	public static final byte[] columf = Bytes.toBytes("f");

	public static final byte[] columTimestamp = Bytes.toBytes("t");
	public final static byte[] columUserId = Bytes.toBytes("u");
	public final static byte[] columDomain = Bytes.toBytes("d");
	public final static byte[] columUrl = Bytes.toBytes("ur");
	public final static byte[] columChannelId = Bytes.toBytes("c");
	public final static byte[] columIp = Bytes.toBytes("i");
	public final static byte[] columbrowser = Bytes.toBytes("b");
	public final static byte[] columCountryId = Bytes.toBytes("t");

	public static byte[] createHbaseKey(long timestamp, int indext) {
		return Bytes.add(Bytes.toBytes(timestamp), Bytes.toBytes(indext));
	}

	public static Put putDataToTable(int indext, DataStruct dataStruct) {
		Put put = new Put(createHbaseKey(dataStruct.getTimestamp(), indext));
		put.add(columf, columTimestamp,
				Bytes.toBytes(dataStruct.getTimestamp()));
		put.add(columf, columUserId, Bytes.toBytes(dataStruct.getUserId()));
		put.add(columf, columDomain, Bytes.toBytes(dataStruct.getDomain()));
		put.add(columf, columUrl, Bytes.toBytes(dataStruct.getUrl()));
		put.add(columf, columChannelId,
				Bytes.toBytes(dataStruct.getChannelId()));
		put.add(columf, columIp, Bytes.toBytes(dataStruct.getIp()));
		put.add(columf, columbrowser, Bytes.toBytes(dataStruct.getBrowser()));
		put.add(columf, columCountryId, Bytes.toBytes(dataStruct.getCountry()));
		return put;
	}

	public static DataStruct getDataFromTable(Result rs) {
		long timeStamp = Bytes.toLong(rs.getValue(columf, columTimestamp));
		int userId = Bytes.toInt(rs.getValue(columf, columUserId));
		String domain = Bytes.toString(rs.getValue(columf, columDomain));
		String url = Bytes.toString(rs.getValue(columf, columUrl));
		int channelId = Bytes.toInt(rs.getValue(columf, columChannelId));
		long ip = Bytes.toInt(rs.getValue(columf, columIp));
		int country = Bytes.toInt(rs.getValue(columf, columCountryId));
		String browser = Bytes.toString(rs.getValue(columf, columbrowser));
		DataStruct dataStruct = new DataStruct(timeStamp, userId, domain, url,
				channelId, ip, browser, country);
		return dataStruct;
	}

	public static String getTableName() {
		return tableName;
	}

	public static HTable htable;

	public static void tryConnect() {
		Configuration conf = HBaseConfiguration.create(new Configuration());
		try {
			htable = new HTable(conf, tableName);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	public static List<DataStruct> scanTable(long startTime, long stopTime) {
		List<DataStruct> result = new ArrayList<DataStruct>();
		tryConnect();
		Scan scan = new Scan();
		scan.setStartRow(createHbaseKey(startTime, 0));
		scan.setStopRow(createHbaseKey(stopTime, Integer.MAX_VALUE));
		try {
			ResultScanner resultScan = htable.getScanner(scan);
			for (Result rs : resultScan) {
				result.add(getDataFromTable(rs));
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return result;
	}
	
	public static void main(String[] args) throws ParseException {
		TrackingTable trackingTable = new TrackingTable();
		SimpleDateFormat simple = new SimpleDateFormat("yyyy-MM-dd:HH-mm");
		long time = simple.parse("2013-03-25:00-02").getTime();
		List<DataStruct> listData = TrackingTable.scanTable(time, time + 10 * 60 * 1000);
		for (DataStruct dt : listData){
			System.out.println(dt);
		}
		System.out.println(listData.size());
	}
}